package com.bds.btcdc.adapter.huobi.market.websokcet;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bds.btcdc.model.CoinPairExchanges;
import com.bds.btcdc.model.MarketDepth;
import com.bds.btcdc.model.Order;
import com.bds.btcdc.api.model.Result;
import com.bds.btcdc.util.Const;
import com.bds.btcdc.util.Utils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.simp.stomp.StompSession;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.text.DecimalFormat;
import java.util.*;
import java.util.zip.GZIPInputStream;

/**
 * MarketDepth 更新推送
 */
public class HuobiDepthReceiver extends WebSocketClient {
	private Logger logger = LoggerFactory.getLogger(getClass());
	private static final DecimalFormat df = new DecimalFormat("#0.00");
	private static final DecimalFormat df2 = new DecimalFormat("#0.0000");

	StompSession session;
	List<CoinPairExchanges> huobi_pairs;

	private static final Map<String,String> bds_pair=new HashMap<>();

	private void initPairs(){
		for (int i = 0; i < huobi_pairs.size(); i++) {
			bds_pair.put(huobi_pairs.get(i).getPair_code(),huobi_pairs.get(i).getEn_pair_code());
		}
	}

	public HuobiDepthReceiver(StompSession session, URI serverURI, List<CoinPairExchanges> huobi_pairs) {
		super(serverURI);
		this.session = session;
		this.huobi_pairs=huobi_pairs;
	}

	public HuobiDepthReceiver(URI serverUri, Draft draft) {
		super(serverUri, draft);
	}

	public HuobiDepthReceiver(URI serverURI) {
		super(serverURI);
	}


	@Override
	public void onOpen(ServerHandshake handshakedata) {
		logger.info("######### HUOBI 连接开始 #############");
		for (int i = 0; i < huobi_pairs.size(); i++) {
			String pair=huobi_pairs.get(i).getPair_code();
			String dataMsg = "{\"sub\":\"market."+pair+".depth.step1\",\"id\":\"id"+i+"\"}";
			send(dataMsg);
		}
		logger.info("######### HUOBI 连接成功 #############");
	}

	@Override
	public void onClose(int code, String reason, boolean remote) {
		logger.info("######### HUOBI连接断开，重连中 ######## 断开原因：" + reason);
		new Thread() {
			@Override
			public void run() {
				try {
					Thread.sleep(Const.RE_CONN_TIME);
					new HuobiDepthReceiver(session, new URI(Const.HUOBI.WS_URL), huobi_pairs).connect();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}.start();
	}

	@Override
	public void onMessage(String message) {
		logger.info(message);
	}

	@Override
	public synchronized void onMessage(ByteBuffer message) {
		//huobi的数据需要解压缩

		ByteArrayOutputStream out = new ByteArrayOutputStream();
		ByteArrayInputStream in = new ByteArrayInputStream(message.array());
		try {
			GZIPInputStream ungzip = new GZIPInputStream(in);
			byte[] buffer = new byte[256];
			int n;
			while ((n = ungzip.read(buffer)) >= 0) {
				out.write(buffer, 0, n);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		byte[] bytes = out.toByteArray();

		try {
		    String real_msg=new String(bytes,"UTF-8");
		    //回复WebSocket Server的心跳
		    if(real_msg.contains("{\"ping\":")){
		        Long time=getPingTime(real_msg);
                send("{\"pong\":"+time+"}");
            }else {
				parseMsg(real_msg);
			}
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
	}

	//解析解压缩后的内容
	private void parseMsg(String real_msg){
		if (bds_pair.size()==0)
			initPairs();
//		logger.info("huobi市场深度更新>>>"+real_msg);

		JSONObject json=JSON.parseObject(real_msg);
		String ch=(String) json.get("ch");
		if(ch==null)
			return;
		String channel=Utils.regex("market\\.(.+)\\.depth",ch);
		String pair=bds_pair.get(channel);
		JSONObject tick=json.getJSONObject("tick");
		JSONArray bids=tick.getJSONArray("bids");
		JSONArray asks=tick.getJSONArray("asks");

		MarketDepth depth=new MarketDepth();
		depth.setExchange(Const.HUOBI.EXCHANGE);
		depth.setPair(pair);
		List<Order> bid_list=new ArrayList<>();
		for (int i = 0; i < 5; i++) {	//深度取到买5
			JSONArray array=bids.getJSONArray(i);
			Order order=new Order();
			order.setPrice(df.format(array.getBigDecimal(0)));
			order.setAmount(df2.format(array.getBigDecimal(1)));
			bid_list.add(order);
		}
		depth.setBids(bid_list);

		List<Order> ask_list=new ArrayList<>();
		for (int i = 0; i < 5; i++) {
			JSONArray array=asks.getJSONArray(i);
			Order order=new Order();
			order.setPrice(df.format(array.getBigDecimal(0)));
			order.setAmount(df2.format(array.getBigDecimal(1)));
			ask_list.add(order);
		}
		depth.setAsks(ask_list);

		Result rst=new Result();
		rst.setCode(Const.SUC);
		rst.setData(depth);
		if (session != null)
			session.send(Const.HUOBI.DEPTH_TOPIC, JSONObject.toJSONString(rst).getBytes());
	}

	@Override
	public void onError(Exception ex) {
		logger.info("an error occurred:" + ex);
	}

	//测试Receiver
	public static void main(String[] args) throws Exception{
		new HuobiDepthReceiver(new URI("wss://api.huobipro.com/ws")).connect();
//		new HuobiDepthReceiver(new URI("wss://api.huobi.br.com/ws")).connect();
	}

	private Long getPingTime(String ping_msg){
        String time_str=Utils.regexAll("\\d+",ping_msg);
        Long time=Long.parseLong(time_str);
        return time;
    }

}